package com.king.first.source;

import com.king.first.bean.Student;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import java.sql.*;

/**
 * https://www.cnblogs.com/laoqing/p/11890929.html Flink 与 mybatis
 */
public class Flink_Source_Mysql {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);

        DataStreamSource<Student> studentDataStreamSource = env.addSource(new RichSourceFunction<Student>() {
            private PreparedStatement ps;
            private Statement statement;
            private Connection conn;

            @Override
            public void open(Configuration parameters) throws Exception {
                System.out.println("init");
                conn = DriverManager.getConnection("jdbc:mysql://hadoop01:3306/testdb?useSSL=false", "root", "j5ry0#jZ7vaUt5f4");
//                ps = conn.prepareStatement("select id, stu_name, age, address, score from student where id in ?");
                statement = conn.createStatement();

            }

            @Override
            public void run(SourceContext<Student> sourceContext) throws Exception {
                System.out.println("query");
                String sql = "insert into person(name, age, createDate) values (?, ?, ?); ";
                String qSql = "select * from person ";
//                boolean execute = statement.execute(qSql);
                ResultSet resultSet = statement.executeQuery(qSql);

                while (resultSet.next()){
                    int anInt = resultSet.getInt("age");
                    String string = resultSet.getString("name");
                    System.out.println(anInt + " - " + string + " - ");
                }
                /*Array anInt = conn.createArrayOf("Int", new Integer[]{1, 2});
                ps.setArray(1, anInt);
                ResultSet resultSet = ps.executeQuery();
                while (resultSet.next()) {
                    Student student = new Student(resultSet.getInt("id"),
                            resultSet.getNString("stu_name"),
                            resultSet.getInt("age"),
                            resultSet.getNString("address"),
                            resultSet.getInt("score")
                    );
                    sourceContext.collect(student);
                }*/
            }

            @Override
            public void close() throws Exception {
                statement.close();
                conn.close();
            }

            @Override
            public void cancel() {

            }
        });
        studentDataStreamSource.print();
        env.execute("mysql");
    }
}

